package com.bw.flinkstreaming.state.job5;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class AggregatingStateWithConcat extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, String>> {

    private AggregatingState<Long, String> totalStr;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建
        AggregatingStateDescriptor<Long, String, String> descriptor =
                new AggregatingStateDescriptor<Long, String, String>(
                        "concatStr",

                        new AggregateFunction<Long, String, String>() {
                            //初始化的操作，只运行一次
                            @Override
                            public String createAccumulator() {
                                return "join：";
                            }

                            // (2,join：4 and 2 and 5)
                            // (1,join：3 and 5 and 7)

                            @Override
                            public String add(Long value, String accumulator) {
                                if ("join：".equals(accumulator)) {
                                    return accumulator + value;
                                }
                                return accumulator + " and " + value;
                            }

                            @Override
                            public String merge(String a, String b) {
                                System.out.println("a+"+ a);
                                System.out.println("b+ "+b);
                                return a + " and " + b;
                            }

                            @Override
                            public String getResult(String accumulator) {
                                return accumulator;
                            }
                        }, String.class); // 状态存储的数据类型
        totalStr = getRuntimeContext().getAggregatingState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, String>> out) throws Exception {
        totalStr.add(element.f1);
        out.collect(Tuple2.of(element.f0, totalStr.get()));
    }
}